package com.fxu.framework.mq.redis.streams;

import com.fxu.framework.core.holder.RedisHolder;
import com.fxu.framework.core.redis.RedisConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import static com.fxu.framework.core.redis.RedisConfig.getJackson2JsonRedisSerializer;

@Slf4j
@Configuration
public class RedisStreamConfig {
    private List<Subscription> subscriptions = new ArrayList<>();

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> streamListenerContainer(
            RedisConnectionFactory factory, List<RedisStreamListener<String, MapRecord<String, Object, Object>>> listeners) {
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = getJackson2JsonRedisSerializer();
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, Object, Object>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .pollTimeout(Duration.ofSeconds(5)) // 拉取消息超时时间
                        .batchSize(10) // 批量抓取消息
                        .targetType(String.class) // 传递的数据类型
                        //.executor(executorService)
                        .keySerializer(new StringRedisSerializer())
                        .hashKeySerializer(new StringRedisSerializer())
                        .hashValueSerializer(new RedisConfig.LZ4Serializer(jackson2JsonRedisSerializer))
                        .build();
        StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> container = StreamMessageListenerContainer
                .create(factory, options);

        /*System.out.println("getKeySerializer=" + options.getKeySerializer());
        System.out.println("getHashKeySerializer=" + options.getHashKeySerializer());
        System.out.println("getHashValueSerializer=" + options.getHashValueSerializer());

        Subscription subscription = container.receive(Consumer.from("consumerGroup", "consumerName"),
                ReadOffset.latest(), listeners.get(0));*/
        log.info("RedisStreamConfig streamListenerContainer={}", listeners.size());
        //注册消费者 消费者名称，从哪条消息开始消费，消费者类
        // > 表示没消费过的消息, $ 表示最新的消息
        listeners.forEach(listener -> {
            String streamKey = listener.getStreamKey();
            String groupName = listener.getGroupName();
            log.info("RedisStreamConfig listener={}, key={}, group={}", listener.getClass(), streamKey, groupName);
            initRedisStream(streamKey, groupName);
            Consumer cunsumer = Consumer.from(groupName, listener.getConsumerName());
            StreamOffset<String> streamOffset = StreamOffset.create(streamKey, ReadOffset.lastConsumed());
            subscriptions.add(container.receive(cunsumer, streamOffset, listener));
        });

        container.start();

        return container;
    }

    public void initRedisStream(String streamKey, String groupName) {
        RedisTemplate<String, Object> redisTemplate = RedisHolder.getRedisTemplate();
        //如果 流不存在 创建 stream 流
        if (Boolean.FALSE.equals(redisTemplate.hasKey(streamKey))) {
            redisTemplate.opsForStream().add(streamKey, Collections.singletonMap("", ""));
            log.info("init redis stream={} success!", streamKey);
        }

        //创建消费者组
        try {
            redisTemplate.opsForStream().createGroup(streamKey, groupName);
        } catch (Exception e) {
            log.info("init redis stream={}, group={} is exists!", streamKey, groupName);
        }
    }
}